Day 21：情報沉澱：將戰情室情資無縫同步至個人第二大腦 (Obsidian Vault 雙鏈知識網)

2026 iThome 鐵人賽

DAY 21

Build on Google AI

打造專屬 AI 參謀群：以 Gemini Spark、Workspace 與 ADK 構建自動化決策雷達系列第 21 篇

Day 21：情報沉澱：將戰情室情資無縫同步至個人第二大腦 (Obsidian Vault 雙鏈知識網)

18th鐵人賽

mervyn67

2026-10-05 23:55:23

38 瀏覽

分享至

在 Day 20 中，我們在 Google Sheets 上構建了兼具 KPI 摘要、動態篩選與卡片閱覽室的戰情室儀表板。然而，試算表的核心定位是「高頻資料的暫存、快速檢視與樞紐決策」，並非適合長期累積技術認知、雙向超鏈結與非線性思考的知識庫。

當一篇論文或研報經過四大代理人深化處理，被標記為 PROCESSED 且評分達到高標（例如 $\ge 4$）時，它就轉化為值得永久保存的個人資產。今天我們將貫徹六角架構，為系統建立一個對外的知識庫轉接器（ObsidianVaultAdapter），將試算表中的精煉情報自動匯出為具備 YAML Frontmatter、雙向鏈接（Wikilinks）與結構化 Callouts 的 Markdown 檔案，沉澱至個人第二大腦（Obsidian Vault）。

一、架構擴展：從單向消耗到資產沉澱

在過去的巡航管線中，資料流向是單向的：
PENDING（生資料）➔ 管線處理 ➔ PROCESSED（寫回 Google Sheets）。

進入情報沉澱階段後，我們需要一個反向提取（Egress）機制。整體架構如下：

+-----------------------------------+
|  Google Sheets (raw_feed)         |
|  - 篩選: PROCESSED & Score >= 4    |
+-----------------------------------+
                  │
                  ▼ (SheetsAdapter.fetch_processed_items)
+-----------------------------------+
|  DataHarmonizer (反序列化與淨化)  |
+-----------------------------------+
                  │ (List[IntelligenceItem])
                  ▼
+-----------------------------------+
|  ObsidianVaultAdapter             |
|  - 防碰撞檔名生成 (Sanitize)      |
|  - Jinja2 模板渲染 Markdown       |
|  - 冪等性寫入 (Idempotent Sync)   |
+-----------------------------------+
                  │
                  ▼
+-----------------------------------+
|  Local Obsidian Vault             |
|  - 00_Inbox/Radar/*.md            |
+-----------------------------------+

二、六角架構端口定義：KnowledgeSyncPort

我們在 src/ports/knowledge_sync.py 定義統一的輸出端口，未來若要擴充支援 Notion API 或 Logseq，只要抽換實作即可：

# src/ports/knowledge_sync.py
from abc import ABC, abstractmethod
from typing import List, Dict
from src.core.models import IntelligenceItem

class KnowledgeSyncPort(ABC):
    @abstractmethod
    def sync_item(self, item: IntelligenceItem, overwrite: bool = False) -> bool:
        """將單筆情報同步至外部知識庫，回傳是否成功寫入（若存在且未覆寫則回傳 False）。"""
        pass

    @abstractmethod
    def batch_sync(self, items: List[IntelligenceItem], overwrite: bool = False) -> Dict[str, int]:
        """批次同步情報清單，回傳成功與略過筆數。"""
        pass

三、適配器實作：ObsidianVaultAdapter

實作此適配器時，需要解決兩個關鍵邊界問題：

檔名安全性與唯一性（防碰撞）：作業系統禁止檔名出現 :*?"<>|/ 等字元；且同一天內若有相同主題的情報，必須附加 entry_id 短碼以防同名覆寫。
屬性容錯與 Markdown 渲染：利用 Jinja2 模板動態組合 YAML 標頭與 Obsidian 特有的 Callout 區塊。

# src/adapters/obsidian_adapter.py
import re
import logging
from pathlib import Path
from typing import List, Dict
from jinja2 import Template
from src.ports.knowledge_sync import KnowledgeSyncPort
from src.core.models import IntelligenceItem

logger = logging.getLogger(__name__)

OBSIDIAN_TEMPLATE = """---
id: "{{ item.entry_id }}"
title: "{{ (item.source_title or '') | replace('"', '\\"') }}"
source: "{{ item.source_url }}"
domain: "{{ item.domain }}"
score: {{ item.verified_score or item.raw_score or 0 }}
date: "{{ item.timestamp if item.timestamp else '' }}"
tags:
  - intelligence-radar
  - domain/{{ item.domain | lower }}
---

# {{ item.source_title }}

> [!abstract] 核心提煉
> {{ item.editorial_summary or "暫無摘要" }}

## 📊 量化對比與效能指標
{{ item.metrics or "無顯著量化指標揭露。" }}

## ⚠️ 技術邊界與潛在缺陷
{{ item.limitations or "作者未於論文中強調顯著邊界。" }}

## 🔮 二階效應與跨領域衝擊
> [!tip] 跨界推演
> {{ item.cross_impact_notes or "暫無推演。" }}

---
## 關聯實體
- 領域節點: [[{{ item.domain }}]]
- 原始追蹤: [論文/研報連結]({{ item.source_url }})
"""

class ObsidianVaultAdapter(KnowledgeSyncPort):
    def __init__(self, vault_path: str | Path, subfolder: str = "00_Inbox/Radar"):
        self.vault_path = Path(vault_path).resolve()
        self.target_dir = self.vault_path / subfolder
        # 防禦性遞迴建立目錄，使用者無需手動預建
        self.target_dir.mkdir(parents=True, exist_ok=True)
        self.template = Template(OBSIDIAN_TEMPLATE)

    @staticmethod
    def _sanitize_filename(name: str) -> str:
        """過濾檔案系統非法字元並限制長度。"""
        sanitized = re.sub(r'[\\/*?:"<>|]', "", name)
        return sanitized.strip()[:80]

    def _build_filepath(self, item: IntelligenceItem) -> Path:
        """產生帶有日期、領域、安全標題與 entry_id 短碼的檔案路徑。"""
        # 日期字串解析
        raw_date = getattr(item, "timestamp", None)
        if hasattr(raw_date, "strftime"):
            date_prefix = raw_date.strftime("%Y-%m-%d")
        elif isinstance(raw_date, str) and len(raw_date) >= 10:
            date_prefix = raw_date[:10]
        else:
            from datetime import datetime
            date_prefix = datetime.now().strftime("%Y-%m-%d")

        safe_title = self._sanitize_filename(item.source_title or "Untitled")
        domain = item.domain or "GENERAL"
        
        # 加上 entry_id 前 8 碼避免同日同題目的檔案碰撞
        entry_short = str(item.entry_id)[:8] if item.entry_id else "no_id"
        filename = f"{date_prefix} - [{domain}] {safe_title} ({entry_short}).md"
        return self.target_dir / filename

    def sync_item(self, item: IntelligenceItem, overwrite: bool = False) -> bool:
        target_file = self._build_filepath(item)
        if target_file.exists() and not overwrite:
            logger.info(f"檔案已存在，略過寫入: {target_file.name}")
            return False

        content = self.template.render(item=item)
        target_file.write_text(content, encoding="utf-8")
        logger.info(f"成功沉澱筆記: {target_file.name}")
        return True

    def batch_sync(self, items: List[IntelligenceItem], overwrite: bool = False) -> Dict[str, int]:
        stats = {"synced": 0, "skipped": 0}
        for item in items:
            if self.sync_item(item, overwrite=overwrite):
                stats["synced"] += 1
            else:
                stats["skipped"] += 1
        return stats

四、實戰踩坑與核心修復記

在實際串聯 Google Sheets 與 Obsidian 的過程中，我們遭遇了 4 個嚴謹的架構與工程挑戰：

1. `DataHarmonizer` 的職責演進：防止「過度淨化」

踩坑現象：成功抓到 2 筆情資，但匯出至 Obsidian 的 Markdown 筆記中，> [!abstract] 核心提煉 卻全是「暫無摘要」，即使 Google Sheets 裡明明填寫了「資深編輯精煉...」。
根本原因：舊版 DataHarmonizer.normalize 是專為「前哨站採集生情資」設計的，內部硬編碼將 status 強制指定為 PENDING，並將審查欄位 verified_score、editorial_summary 通通重設為 None。
重構修復：擴充 DataHarmonizer，使其兼具「採集時賦予預設值」與「反序列化時完整保留下遊審查成果」的健全雙向能力：

# src/core/harmonizer.py (節錄)
class DataHarmonizer:
    @staticmethod
    def normalize(raw_payload: Dict[str, Any]) -> IntelligenceItem:
        entry_id = raw_payload.get("entry_id") or str(uuid.uuid4())[:8]
        timestamp = DataHarmonizer._parse_timestamp(raw_payload.get("timestamp"))

        # 保留既有狀態；無狀態時才預設為 PENDING
        raw_status = raw_payload.get("status")
        try:
            status = ItemStatus(raw_status) if raw_status else ItemStatus.PENDING
        except ValueError:
            status = ItemStatus.PENDING

        # 容錯讀取審查評分
        raw_verified = raw_payload.get("verified_score")
        try:
            verified_score = int(raw_verified) if raw_verified not in (None, "") else None
        except (ValueError, TypeError):
            verified_score = None

        # 核心欄位白名單與 metadata 自動打包
        core_fields = {
            "entry_id", "timestamp", "status", "domain", "track",
            "source_title", "source_url", "raw_content", "metrics",
            "limitations", "raw_score", "verified_score",
            "editorial_summary", "cross_impact_notes", "metadata"
        }
        merged_metadata = dict(raw_payload.get("metadata", {}))
        for k, v in raw_payload.items():
            if k not in core_fields and v != "":
                merged_metadata[k] = v

        return IntelligenceItem(
            entry_id=str(entry_id),
            timestamp=timestamp,
            status=status,
            domain=raw_payload.get("domain", "GENERAL"),
            track=raw_payload.get("track", "DEFAULT"),
            source_title=str(raw_payload.get("source_title", "Untitled")).strip(),
            source_url=str(raw_payload.get("source_url", "")).strip(),
            raw_content=str(raw_payload.get("raw_content", "")).strip(),
            metrics=raw_payload.get("metrics") or None,
            limitations=raw_payload.get("limitations") or None,
            raw_score=int(raw_payload.get("raw_score", 3)),
            verified_score=verified_score,
            editorial_summary=raw_payload.get("editorial_summary") or None,
            cross_impact_notes=raw_payload.get("cross_impact_notes") or None,
            metadata=merged_metadata
        )

2. Google Sheets 欄位對齊與 `gspread` 物件階層

在 SheetsAdapter 中新增 fetch_processed_items 時需注意：

gc.open_by_key(...) 取得的是 Spreadsheet 物件，而 get_all_records() 是定義在分頁 Worksheet（即 self.worksheet）之上。
評分欄位不是普通的 score，而是經由 Reviewer 認證的 verified_score。

# src/adapters/sheets_adapter.py (節錄)
def fetch_processed_items(self, min_score: int = 4) -> List[IntelligenceItem]:
    """從 raw_feed 提取已處理且評分達到門檻的情報。"""
    records = self.worksheet.get_all_records()
    processed: List[IntelligenceItem] = []

    for record in records:
        if str(record.get("status", "")).strip().upper() != ItemStatus.PROCESSED.value:
            continue

        raw_score = record.get("verified_score") or record.get("raw_score") or 0
        try:
            score = float(raw_score)
        except (ValueError, TypeError):
            score = 0.0

        if score >= min_score:
            # 統一透過 Harmonizer 進行反序列化與淨化
            item = DataHarmonizer.normalize(record)
            processed.append(item)

    return processed

3. 檔名碰撞與冪等性略過（Idempotency Collision）

踩坑現象：執行同步時輸出 [+] 同步完成！新增: 1 篇，略過: 1 篇，明明有兩筆不同的紀錄。
原因：兩筆測試資料的發布日期、分類與標題剛好相同，產生的目標檔名完全一致。轉接器為了防止重複覆寫，判定第二筆「檔案已存在」而跳過。
修復：在 _build_filepath 中強制附加 (entry_id[:8]) 短碼，確保每筆實體在檔案系統中具備唯一空間。

4. Python 模組執行與 Logging 單一進入點規範

當腳本放在與 src/ 平行的 scripts/sync_to_obsidian.py 時，直接以 python scripts/sync_to_obsidian.py 執行會因為 sys.path 找不到 src 而拋出 ModuleNotFoundError。標準規範是在專案根目錄使用模組化執行：

python -m scripts.sync_to_obsidian

模組內部（如 sheets_adapter.py）只宣告 logger = logging.getLogger(__name__)，維持純淨；全域的 logging.basicConfig(...) 嚴格僅在 scripts/sync_to_obsidian.py 等主入口初始化一次。

五、同步腳本撰寫與本機驗證

在 scripts/sync_to_obsidian.py 編寫排程或手動執行的 CLI 工具：

# scripts/sync_to_obsidian.py
import os
import logging
from pathlib import Path
from src.adapters.sheets_adapter import SheetsAdapter
from src.adapters.obsidian_adapter import ObsidianVaultAdapter

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - [%(levelname)s] - %(message)s"
)
logger = logging.getLogger(__name__)

def main():
    # 預設路徑可透過環境變數 OBSIDIAN_VAULT_PATH 覆寫
    vault_path_env = os.getenv("OBSIDIAN_VAULT_PATH", "~/ObsidianVault")
    vault_path = Path(vault_path_env).expanduser()

    logger.info(f"初始化 Obsidian 轉接器，目標 Vault: {vault_path}")
    obsidian_adapter = ObsidianVaultAdapter(
        vault_path=vault_path,
        subfolder="00_Inbox/Radar"
    )

    logger.info("連線至 Google Sheets 讀取已處理情報...")
    sheets_adapter = SheetsAdapter()
    candidates = sheets_adapter.fetch_processed_items(min_score=4)
    logger.info(f"[*] 找到 {len(candidates)} 筆高價值情報候選項目...")

    results = obsidian_adapter.batch_sync(candidates)
    logger.info(f"[+] 同步完成！新增: {results['synced']} 篇，略過: {results['skipped']} 篇。")

if __name__ == "__main__":
    main()

執行成果

在專案終端機執行：

python -m scripts.sync_to_obsidian

輸出訊息：

2026-10-06 22:45:10 - [INFO] - 初始化 Obsidian 轉接器，目標 Vault: /Users/mervyn/ObsidianVault
2026-10-06 22:45:11 - [INFO] - 連線至 Google Sheets 讀取已處理情報...
2026-10-06 22:45:12 - [INFO] - [*] 找到 2 筆高價值情報候選項目...
2026-10-06 22:45:12 - [INFO] - 成功沉澱筆記: 2026-09-25 - [SCIENCE] State Locking Verification Paper (dfd9abe0).md
2026-10-06 22:45:12 - [INFO] - 成功沉澱筆記: 2026-09-25 - [SCIENCE] State Locking Verification Paper (ddad0487).md
2026-10-06 22:45:12 - [INFO] - [+] 同步完成！新增: 2 篇，略過: 0 篇。

打開 Obsidian 檔案夾，產出的 Markdown 內容兼備 Frontmatter 與結構化 Callouts：

---
id: "dfd9abe0"
title: "State Locking Verification Paper"
source: "https://arxiv.org/abs/test.lock"
domain: "SCIENCE"
score: 5
date: "2026-09-25 14:50:55"
tags:
  - intelligence-radar
  - domain/science
---

# State Locking Verification Paper

> [!abstract] 核心提煉
> 資深編輯審核：確認狀態機在多進程並發下具備完整鎖定能力，能有效避免重入攻擊與重複處理。

## 📊 量化對比與效能指標
無顯著量化指標揭露。

## ⚠️ 技術邊界與潛在缺陷
作者未於論文中強調顯著邊界。

## 🔮 二階效應與跨領域衝擊
> [!tip] 跨界推演
> 暫無推演。

---
## 關聯實體
- 領域節點: [[SCIENCE]]
- 原始追蹤: [論文/研報連結](https://arxiv.org/abs/test.lock)

至此，我們正式打破了「雲端試算表」與「本機個人知識庫」之間的壁壘，流動情資至此沉澱為可供非線性思維鏈接的長效第二大腦資產。